package org.ala.tiktools.rocketmq.producer;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rocketmq.MQMessage;
import org.ala.tiktools.rocketmq.SnowflakeIdGenerator;
import org.ala.tiktools.rocketmq.producer.queue.IQueueSelecter;
import org.ala.tiktools.rocketmq.producer.transaction.DefaultTransactionListener;
import org.ala.tiktools.rocketmq.serializable.kryo.KryoTool;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.ClassUtils;


/**
 * producer执行器
 *
 * @author ala
 * @date 2024-12-05 11:57
 */
public class ProducerExecutor implements ApplicationContextAware {


    static Logger log = LoggerFactory.getLogger("rocketmq", "producer");


    /**
     *  namesrcAddrs
     */
    protected String namesrcAddrs;
    /**
     *  group, topic, tag
     */
    protected String group, topic, tag;
    /**
     *  默认的队列数量
     */
    protected int defaultQueueNum;
    /**
     *  队列选择器
     */
    protected IQueueSelecter selector;
    protected Class<? extends IQueueSelecter> selectorClass;
    /**
     *  事务消息
     */
    protected TransactionListener transactionListener;
    protected Class<? extends TransactionListener> transactionListenerClass;

    /**
     *
     */
    protected DefaultMQProducer producer;

    /**
     *  id生成器
     */
    protected SnowflakeIdGenerator idGenerator;


    public ProducerExecutor(String namesrcAddrs, String group, String topic, String tag, int defaultQueueNum,
                            Class<? extends IQueueSelecter> selectorClass,
                            Class<? extends TransactionListener> transactionListenerClass,
                            ApplicationContext applicationContext) {
        this.namesrcAddrs = namesrcAddrs;
        this.group = group;
        this.topic = topic;
        this.tag = tag;
        this.defaultQueueNum = defaultQueueNum;
        this.selectorClass = selectorClass;
        this.transactionListenerClass = transactionListenerClass;
        this.setApplicationContext(applicationContext);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //  初始化队列选择器
        if (selectorClass != null) {
            try {
                selector = applicationContext.getBean(selectorClass);
            } catch (Exception e) {
                try {
                    selector = selectorClass.newInstance();
                } catch (InstantiationException | IllegalAccessException ex) {
                    log.error("[ProducerExecutor setApplicationContext] 初始化queueSelector失败。找不到queueSelector类型的bean，new也异常。请确认是否包含无参构造函数. class:" + selectorClass, e);
                    throw new RuntimeException(ex);
                }
            }
        }
        //  初始化事务监听
        if (transactionListenerClass != null) {
            try {
                transactionListener = applicationContext.getBean(transactionListenerClass);
            } catch (Exception e) {
                try {
                    transactionListener = transactionListenerClass.newInstance();
                } catch (InstantiationException | IllegalAccessException ex) {
                    log.error("[ProducerExecutor setApplicationContext] 初始化TransactionListener失败。找不到TransactionListener类型的bean，new也异常。请确认是否包含无参构造函数. class:" + transactionListenerClass, e);
                    throw new RuntimeException(ex);
                }
            }
        }

        //  初始化id生成器
        idGenerator = applicationContext.getBean("RocketMQSnowflakeIdGenerator", SnowflakeIdGenerator.class);

        //  初始化producer
        try {
            initProducer();
        } catch (MQClientException e) {
            log.error("[ProducerExecutor setApplicationContext] 启动Producer失败。请确认地址信息是否正确, namesrvAddrs:" + namesrcAddrs + " group:" + group + " topic:" + topic, e);
            throw new RuntimeException(e);
        }
    }
    /**
     *  初始化Producer
     */
    protected void initProducer() throws MQClientException {
        DefaultMQProducer producer = null;
        if (transactionListener != null) {
            TransactionMQProducer tx = new TransactionMQProducer(group);
            tx.setTransactionListener(transactionListener);
            producer = tx;
        } else {
            producer = new DefaultMQProducer(group);
        }

        if (defaultQueueNum >= 0) {
            producer.setDefaultTopicQueueNums(defaultQueueNum);
        }

        producer.setNamesrvAddr(namesrcAddrs);
        producer.start();
        this.producer = producer;

        log.info("[ProducerExecutor initProducer] 开启Producer推送 namesrc:" + namesrcAddrs + " group:" + group + " topic:" + topic);
    }


    /**
     *  发送消息
     */
    public void send(Object[] args, int tagIdx) throws Exception {
        MessageExt msg = new MessageExt();
        msg.setTopic(topic);
        //  给tag
        msg.setTags(tag);
        if (tagIdx >= 0) {
            msg.setTags(args[tagIdx].toString());
        }
        //  给消息体
        MQMessage mqMessage = new MQMessage();
        mqMessage.setId(idGenerator.id());
        mqMessage.setSendTimestamp(System.currentTimeMillis());
        mqMessage.setBody(args);
        msg.setBody(KryoTool.getInstance().serialize(mqMessage));
        if (selector != null) {
            msg.setQueueId(selector.select(args, producer.getDefaultTopicQueueNums()));
        }
        producer.send(msg);

        log.info("[ProducerExecutor send] 发送消息. group:" + group + " topic:" + topic + " tag:" + tag + " id:" + mqMessage.getId());
    }


    public String getNamesrcAddrs() {
        return namesrcAddrs;
    }

    public void setNamesrcAddrs(String namesrcAddrs) {
        this.namesrcAddrs = namesrcAddrs;
    }

    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public int getDefaultQueueNum() {
        return defaultQueueNum;
    }

    public void setDefaultQueueNum(int defaultQueueNum) {
        this.defaultQueueNum = defaultQueueNum;
    }

    public String getTag() {
        return tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }

    public Class<? extends IQueueSelecter> getSelectorClass() {
        return selectorClass;
    }

    public void setSelectorClass(Class<? extends IQueueSelecter> selectorClass) {
        this.selectorClass = selectorClass;
    }

    public Class<? extends TransactionListener> getTransactionListenerClass() {
        return transactionListenerClass;
    }

    public void setTransactionListenerClass(Class<? extends TransactionListener> transactionListenerClass) {
        this.transactionListenerClass = transactionListenerClass;
    }
}
